package org.ykhl.mq.commons.pusher;

import com.google.common.base.Strings;
import org.ykhl.mq.commons.Constants;
import org.ykhl.mq.commons.Utils;
import org.ykhl.mq.commons.BaseMessage;
import org.ykhl.mq.commons.ConnectionException;

import java.util.concurrent.Future;

/**
 * 消息推送类的抽象父类，用于维护一些基本的生命周期管理逻辑和通用的成员变量等。
 * 
 * @author Zhang Kefei
 *
 */
public abstract class MessagePusher {
	// the id of hospital or app server
	protected String authKey;
	protected Constants.Role role;
	protected String producerGroupName;
	protected String topic;
	protected Boolean useSsl;
	protected String securityFilePath;
	
	// the server address of MQ
	protected String serverAddress;
	
	private boolean running = false;

	protected MessagePusher(String serverAddress, String authKey,
							String securityFilePath, Constants.Role role){

		init(serverAddress, authKey, securityFilePath, role,
				Constants.MQ_FROM_APPSERVER_TOPIC_PREFIX,
				Constants.MQ_FROM_APPSERVER_PRODUCER_PREFIX);

	}

	protected MessagePusher(String serverAddress, String authKey,
							String securityFilePath, Constants.Role role,
							String topicPrefix, String producerPrefix){
		init(serverAddress, authKey, securityFilePath, role, topicPrefix, producerPrefix);
	}

	private void init(String serverAddress, String authKey, String securityFilePath, Constants.Role role,
							String topicPrefix, String producerPrefix){
		if (Strings.isNullOrEmpty(authKey)) {
			if (Strings.isNullOrEmpty(securityFilePath)) {
				throw new NullPointerException(
						"either authKey or securityFilePath should be non-null");
			}

			authKey = Utils.getAuthIdFromKeyStoreUnchecked(securityFilePath);
		}

		if(Strings.isNullOrEmpty(serverAddress) || Strings.isNullOrEmpty(authKey))
			throw new NullPointerException("serverAddress or authKey is empty");

		this.serverAddress = serverAddress;
		this.authKey = authKey;
		this.securityFilePath = Strings.emptyToNull(securityFilePath);
		this.role = role;
		topic = topicPrefix + authKey;
		producerGroupName = producerPrefix + authKey;
	}
	
	public void setServerAddress(String serverAddress){
		if(serverAddress == null)
			throw new NullPointerException();
		
		this.serverAddress = serverAddress;
	}

	/**
	 * 设置是否使用加密及认证模式进行传输。
	 *
	 * @param useSsl
     */
	public void setUseSsl(Boolean useSsl) {
		this.useSsl = useSsl;
	}
	
	public boolean isRunning(){
		return running;
	}
	
	/**
	 * 用于启动消息推送器的方法。
	 * 
	 * @throws ConnectionException
	 */
	public void start() throws ConnectionException {
		if(running)
			return;
		
		try{
			doStart();
			
			running = true;
		} catch(Exception e){
			throw new ConnectionException(e);
		}
	}
	
	/**
	 * 用于停止消息推送器的方法。
	 * 
	 * @throws ConnectionException
	 */
	public void stop() throws ConnectionException {
		if(!running)
			return;
		
		try{
			doStop();
			
			running = false;
		} catch(Exception e){
			throw new ConnectionException(e);
		}
	}
	
	/**
	 * 向消息安全平台发送推送消息。 使用前请确保已经调用过start方法。
	 *
	 * @param msg 消息体
	 * @return true 发送成功， false 发送失败
	 * @throws ConnectionException
	 */
	public boolean sendMessage(PushMessage msg) throws ConnectionException {
		if(!running)
			throw new IllegalStateException("MessagePusher not started yet");
		if(msg == null)
			throw new NullPointerException();
		
		try{
			return doSendMessage(msg);
		} catch(Exception e){
			throw new ConnectionException(e);
		}
	}

	/**
	 * 向消息安全平台发送推送消息。 使用前请确保已经调用过start方法。
	 *
	 * @param msg 消息体
	 * @return 与发送消息绑定的Future
	 * @throws ConnectionException
     */
	public Future<?> sendMessageAsync(PushMessage msg) throws ConnectionException {
		if(!running)
			throw new IllegalStateException("MessagePusher not started yet");
		if(msg == null)
			throw new NullPointerException();

		try {
			return doSendMessageAsync(msg);
		} catch(Exception e){
			throw new ConnectionException(e);
		}
	}

	protected abstract Future<?> doSendMessageAsync(PushMessage msg);
	protected abstract void doStart() throws Exception;
	protected abstract void doStop() throws Exception;
	protected abstract boolean doSendMessage(PushMessage pushMessage) throws Exception;
	protected abstract BaseMessage trans2Message(PushMessage pushMessage);
}
